package com.reload.reaactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.time.Duration;

/**
 * @Date: 2024/12/5 14:07
 * @Author: jianchenlong
 * @Description: 事件感知
 */
public class EventAware {


    /**
     * Flux事件感知
     * 响应式编程核心：看懂文档弹珠图
     * SignalType：
     * SUBSCRIBE： 被订阅
     * REQUEST：  请求了N个元素
     * CANCEL： 流被取消
     * ON_SUBSCRIBE：在订阅时候
     * ON_NEXT： 在元素到达
     * ON_ERROR： 在流错误
     * ON_COMPLETE：在流正常完成时
     * AFTER_TERMINATE：中断以后
     * CURRENT_CONTEXT：当前上下文
     * ON_CONTEXT：感知上下文
     * doOnXxx API触发时机
     * 1、doOnNext：每个数据（流的数据）到达的时候触发
     * 2、doOnEach：每个元素（流的数据和信号）到达的时候触发
     * 3、doOnRequest： 消费者请求流元素的时候
     * 4、doOnError：流发生错误
     * 5、doOnSubscribe: 流被订阅的时候
     * 6、doOnTerminate： 发送取消/异常信号中断了流
     * 7、doOnCancle： 流被取消
     * 8、doOnDiscard：流中元素被忽略的时候
     */
    public static void main(String[] args) throws Exception {
        // doOnXxxx 事件感知，发生什么事的时候触发什么事
        Flux<Object> empty = Flux.empty()
                .doOnComplete(() -> System.out.println("流正常结束"));
        empty.subscribe(System.out::println);

        // 链式API中，下面的操作符操作的是上面的流
        // Flux<Integer> integer = Flux.just(1, 2, 3, 4,5,6,7)
        Flux<Integer> integer = Flux.range(1, 7)
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> System.out.println("发布者-流正常结束"))
                // 订阅者取消时执行
                .doOnCancel(() -> System.out.println("发布者-流被取消"))
                .doOnError(e -> System.out.println("发布者-流发生错误" + e))
                .doOnRequest(n -> System.out.println("发布者-流被请求" + n + "个元素"))
                // 元素达到时候触发
                .doOnNext(i -> System.out.println("发布者-doOnNext = " + i))
                // each封装的信息比next更加多
                .doOnEach(signal -> System.out.println("发布者-信号类型" + signal.getType()));

        integer.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅者和发布者绑定好了：" + subscription);
                subscription.request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("订阅 元素达到 = " + value);
                if (value < 5) {
                    request(1);
                    if (value == 3) {
                        // throw new RuntimeException("手动抛出异常");
                        cancel();
                    }
                } else {
                    cancel();
                }
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("订阅者-数据流结束");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("订阅者-数据流发生错误" + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("数据流被取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                super.hookFinally(type);
                System.out.println("数据流结束，类型为：" + type);
            }
        });
        System.in.read();
    }
}
